package me.xhy.java.springboot3.webflux.projectreactor.s2java9flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * @author xhy
 * @since 2024-10-30 上午8:29
 */
public class S1Flow {

  public static void main(String[] args) throws InterruptedException {
    // JVM 已经将发布定于模式设计成： 缓冲区 + 多线程 的异步形式
    b();

    // 订阅默认使用 ForkJoinPool 线程池，需要保持主线程存在
    TimeUnit.SECONDS.sleep(100);
  }

  public static void a() {
    /*
     * Java 9 增加了新特性 FLow， 与响应式编程相关。 内部来定义了以下对象
     * Publisher 数据发布者
     * Subscriber 数据订阅者
     * Subscription 订阅关系
     */

    // Java 9 和 Spring Flux 都有 Publisher 类，这里使用 Java 9 的。
    // 定义一个数据发布者，发布String类型的数据
    Flow.Publisher<String> publisher = new Flow.Publisher<String>() {

      // 订阅，参数是Subscriber，将Subscriber传递给Publisher。
      @Override
      public void subscribe(Flow.Subscriber<? super String> subscriber) {
        // Publisher应该将Subscriber保存起来，知道谁对数据感兴趣。
        // this.subscriber = subscriber;
      }
    };

  }

  public static void b() throws InterruptedException {
    // 1. 一个简单直观的Publisher
    SubmissionPublisher submissionPublisher = new SubmissionPublisher();

    // 2. Subscriber
    Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

      // 订阅关系
      Flow.Subscription subscription;

      // onXxx() 方法，在xxx事件发生时，执行这个回调

      // 在订阅时回调，会将订阅关系传递过来
      @Override
      public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("onSubscribe: " + subscription + "，线程: " + Thread.currentThread());

        // 保存订阅关系
        this.subscription = subscription;
        // 在订阅时，request 数据，触发订阅者的 onNext
        this.subscription.request(1);
      }

      // 在下一个元素到达时=接收到数据
      @Override
      public void onNext(String item) {
        System.out.println("onNext: " + item + "，线程: " + Thread.currentThread());

        this.subscription.request(1);

        if (item.equals("publish:10")) {
          // 取消订阅关系
          this.subscription.cancel();
        }
      }

      // 在错误发生时
      @Override
      public void onError(Throwable throwable) {
        System.out.println("onError: " + throwable + "，线程: " + Thread.currentThread());
      }

      @Override
      public void onComplete() {
        System.out.println("onComplete: " + Thread.currentThread() + "，线程: " + Thread.currentThread());
      }
    };

    // 3. 绑定发布者和订阅者
    // Publisher 将所有绑定的 Subscriber 都保存在自己的缓存中，有数据需要发送时，遍历数组，逐一发送数据
    submissionPublisher.subscribe(subscriber);

    // 先完成订阅，再发送数据。订阅关系创建完成前的事件，将会丢失
    for (int i = 0; i < 10; i++) {
      // submit()发布数据
      // SubmissionPublisher 内部的数组对象，缓存着submit()的数据。并且在存入数组之前，有锁控制 抑制并发的发生。
      submissionPublisher.submit("publish:" + i);
      System.out.println("发布数据的线程: " + Thread.currentThread());
    }

    // 不要马上报错
    TimeUnit.SECONDS.sleep(3);
    // closeExceptionally 会插队，没有接收到的 事件（消息事件、完成事件） 也不回到达 Subscriber 了
    submissionPublisher.closeExceptionally(new Throwable("发生错误"));

    // 错误事件发生后，继续发数据，将抛出异常
    TimeUnit.SECONDS.sleep(3);
    for (int i = 10; i < 20; i++) {
      // submit()发布数据
      // SubmissionPublisher 内部的数组对象，缓存着submit()的数据。并且在存入数组之前，有锁控制 抑制并发的发生。
      submissionPublisher.submit("publish:" + i);
      System.out.println("发布数据的线程: " + Thread.currentThread());
    }

    // 关闭发布者
    submissionPublisher.close();

  }


}
